package com.zwh.rocketmq.demo.batch;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.ListUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 批量发送消息
 * @author zwh
 * @date 2024-1-22 9:39
 */
@Service
public class BatchMessageProduct {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /*
      发送批量消息
     */
    public void sendBatchMsg() {
        String destination = "TEST_BATCH_TOPIC";
        String tags = "tag1";
//        String msgBody = "hello rocketMQ";
//        String keys = "orderId1";
        if (StringUtils.isNotBlank(tags)) {
            // 带tag的消息，直接在topic后面加上":tag"， 底层会进行自动转换
            destination = destination + ":" + tags;
        }

        List<Message> messages = new ArrayList<>(100);
        for(int i=0; i<100; i++) {
            Map<String, Object> msgData1 = new HashMap<>(2);
            msgData1.put("operator", "订单生成");
            msgData1.put("orderId", i);
            Message<Map<String, Object>> message1 = MessageBuilder.withPayload(msgData1)

                    .setHeader(RocketMQHeaders.KEYS, i)
                    .build();
            messages.add(message1);
        }

        for(int i = 0; i<100;) {

            List<Message> messageList = messages.subList(i, i + 10 - 1);
            // 官网提示：如果批量消息大于1MB就不要用一个批次发送,拆分成多个批次发送
            // 默认限制时不能超过4MB，否则会程序报错
            rocketMQTemplate.syncSend(destination, messageList);
            i = i + 10;
        }


    }
}
